package jmine.tec.subscription.replicator.jms;

import java.io.Serializable;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import jmine.tec.subscription.Publication;
import jmine.tec.subscription.Publisher;
import jmine.tec.subscription.SubscriptionManager;
import jmine.tec.subscription.SubscriptionSettings;
import jmine.tec.subscription.impl.PartialPublication;
import jmine.tec.subscription.replicator.AbstractIdentifiedReplicator;
import jmine.tec.subscription.replicator.AbstractPublicationWrapper;
import jmine.tec.subscription.replicator.PublicationReplicator;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

/**
 * @author takeshi
 */
public class TopicPublicationReplicator extends AbstractIdentifiedReplicator implements PublicationReplicator, InitializingBean,
        DisposableBean {

    private TopicConnectionFactory topicConnectionFactory;

    private Topic topic;

    private TopicConnection topicConnection;

    private TopicSession topicSession;

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void synchronizePartialPublication(Publisher<P> publisher, SubscriptionSettings settings,
            PartialPublication<P> publication, SubscriptionManager source) {
        try {
            this.sendMessage(this.newPartialPublicationWrapper(publisher, settings, publication));
        } catch (JMSException e) {
            this.handleJMSException(e);
        }
    }

    /**
     * @param e the {@link JMSException}
     */
    private void handleJMSException(JMSException e) {
        LOGGER.warn("JMSException", e);
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void synchronizePublication(Publisher<P> publisher, SubscriptionSettings settings, P publication,
            SubscriptionManager source) {
        try {
            this.sendMessage(this.newPublicationWrapper(publisher, settings, publication));
        } catch (JMSException e) {
            this.handleJMSException(e);
        }
    }

    /**
     * Envia a mensagem via Topic
     * 
     * @param toSend mensagem a ser enviada
     * @throws JMSException e
     */
    private void sendMessage(Serializable toSend) throws JMSException {
        TopicPublisher topicPublisher = this.topicSession.createPublisher(this.topic);
        ObjectMessage message = this.topicSession.createObjectMessage(toSend);
        message.setStringProperty("sentby", this.getIdentification());
        topicPublisher.publish(message);
    }

    /**
     * @param topicConnectionFactory the topicConnectionFactory to set
     */
    public void setTopicConnectionFactory(TopicConnectionFactory topicConnectionFactory) {
        this.topicConnectionFactory = topicConnectionFactory;
    }

    /**
     * @param topic the topic to set
     */
    public void setTopic(Topic topic) {
        this.topic = topic;
    }

    /**
     * {@inheritDoc}
     */
    public void afterPropertiesSet() throws Exception {
        this.topicConnection = this.topicConnectionFactory.createTopicConnection();
        this.topicSession = this.topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
        TopicSubscriber subscriber = this.topicSession.createSubscriber(this.topic, "", false);
        subscriber.setMessageListener(new PublicationMessageListener());
        this.topicConnection.start();
    }

    /**
     * {@inheritDoc}
     */
    public void destroy() throws Exception {
        this.topicSession.close();
        this.topicConnection.close();
    }

    /**
     * {@link MessageListener} que gerencia as mensagens. Usa o resultado de {@link TopicPublicationReplicator#getIdentification()} para
     * verificar a origem da mensagem
     * 
     * @author takeshi
     */
    private class PublicationMessageListener implements MessageListener {

        /**
         * {@inheritDoc}
         */
        public void onMessage(Message arg0) {
            try {
                if (!TopicPublicationReplicator.this.getIdentification().equals(arg0.getStringProperty("sentby"))) {
                    if (arg0 instanceof ObjectMessage) {
                        AbstractPublicationWrapper wrapper = (AbstractPublicationWrapper) ((ObjectMessage) arg0).getObject();
                        TopicPublicationReplicator.this.handle(wrapper);
                    }
                }
            } catch (JMSException e) {
                TopicPublicationReplicator.this.handleJMSException(e);
            }
        }

    }

}
